package com.hhy.flink;

import com.mysql.jdbc.Connection;
import com.mysql.jdbc.PreparedStatement;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import scala.Tuple2;
import scala.Tuple3;

import java.sql.DriverManager;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;


public class ReloadData extends RichMapFunction<Tuple3<String, String,String>, Tuple3<String, String, String>> {

    //定义一个变量，用于保存维表数据在内存
    Map<String, String> dim;

    private Connection connect = null;
    private PreparedStatement ps = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        connect = (Connection) DriverManager.getConnection("jdbc:mysql://10.254.32.7:3306", "root", "root");
        ps = (PreparedStatement) connect.prepareStatement("select mac,apname from bigdata.apmac");
        ResultSet resultSet = ps.executeQuery();
        dim = new HashMap<>();

        while (resultSet.next()) {
            dim.put(resultSet.getString(1).replace(".",""),resultSet.getString(2).replace(".",""));
        }


    }

    @Override
    public Tuple3<String, String, String> map(Tuple3<String, String, String> stringStringStringTuple3) throws Exception {

        return new Tuple3<>(stringStringStringTuple3._1(), dim.get(stringStringStringTuple3._2().replace(":","")), dim.get(stringStringStringTuple3._3().replace(":","")));
    }
}
